package org.example.flinksql.test;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableEnvironment;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;

/**
 * 准备数据
 */
public class CreateData {
    public static void main(String[] args) {
//        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();

        TableEnvironment tbEnv = TableEnvironment.create(bsSettings);

        // CREATE TEMPORARY TABLE t_users
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_users (" +
                "id INT," +
                "username STRING," +
                "password STRING," +
                "nickname STRING," +
                "sex STRING," +
                "age INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_users'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // CREATE TEMPORARY TABLE t_enterprise
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_enterprise (" +
                "eid STRING," +
                "name STRING," +
                "credit_no STRING," +
                "category_new INT," +
                "regist_capi INT," +
                "actual_capi INT," +
                "status INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_enterprise'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        // CREATE TEMPORARY TABLE t_address
        tbEnv.executeSql("CREATE TEMPORARY TABLE t_address (" +
                "eid STRING," +
                "address STRING," +
                "postcode STRING," +
                "u_tags INT) WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
                "'table-name' = 't_address'," +
                "'username' = 'root'," +
                "'password' = '123456'," +
                "'sink.buffer-flush.max-rows' = '0'," +
                "'sink.buffer-flush.interval' = '0'," +
                "'sink.max-retries' = '3'" +
                ")");

        for(int i = 0; i < 8; i ++) {
            tbEnv.executeSql("CREATE TABLE t_enterprise_" + i +" (" +
                    "eid STRING," +
                    "name STRING," +
                    "credit_no STRING," +
                    "category_new INT," +
                    "regist_capi INT," +
                    "actual_capi INT," +
                    "status INT) WITH (" +
                    "'connector' = 'jdbc'," +
                    "'url' = 'jdbc:mysql://127.0.0.1:3306/db-test?useSSL=false'," +
                    "'table-name' = 't_enterprise_" + i + "'," +
                    "'username' = 'flinksql'," +
                    "'password' = 'w0c53w0c53'," +
                    "'sink.buffer-flush.max-rows' = '0'," +
                    "'sink.buffer-flush.interval' = '0'," +
                    "'sink.max-retries' = '3'" +
                    ")");
        }

        // tbEnv.executeSql 只能执行单次Insert任务，速度慢
//        for (int i = 1; i <= 60000; i++) {
//            String username = "user-" + i;
//            String password = "passwd-" + i;
//            String nickname = "nick-" + i;
//            String sex = i % 2 == 1 ? "男" : "女";
//            int age = i % 10;
//            tbEnv.executeSql("insert into t_users(id,username,password,nickname,sex,age) " +
//                    "values(" + i + ",'" + username + "','" + password + "','" + nickname + "','" + sex + "'," + age + ")");
//        }

        // StatementSet可共用
        // statementSet.execute()执行后会清空添加的Insert任务,因此可以共用
        StatementSet statementSet = tbEnv.createStatementSet();

        // statementSet可以批量添加Insert任务
        // statementSet.execute()进行批量提交，速度快
        for (int i = 0; i < 1000; i++) {
            boolean submit = i % 10 == 9 ? true : false;

            String username = "user-" + i;
            String password = "passwd-" + i;
            String nickname = "nick-" + i;
            String sex = i % 2 == 1 ? "男" : "女";
            int age = i % 10;
            statementSet.addInsertSql("insert into t_users(id,username,password,nickname,sex,age) " +
                    "values(" + i + ",'" + username + "','" + password + "','" + nickname + "','" + sex + "'," + age + ")");

            if (submit) {
                statementSet.execute();
            }
        }

        // t_enterprise_0~7
        for (int i = 0; i < 1; i ++) {
            int category_new = i % 2 == 1 ? 115601 : 115602;
            int regist_capi = 100 + i % 100;
            int actual_capi = regist_capi / 2;
            int status = i % 10 == 1 ? 1 : 0;

            for (int j = 0; j < 8; j ++) {
                String eid = UUID.randomUUID().toString() + "_" + j;
                String name = "name-" + eid;
                String credit_no = "credit-" + eid;

                statementSet.addInsertSql("INSERT INTO t_enterprise_" + j +
                        "(eid,name,credit_no,category_new,regist_capi,actual_capi,status)" +
                        " values('" + eid + "','" + name + "','" + credit_no + "'," + category_new + "," +
                        regist_capi + "," + actual_capi + "," + status + ")");
//                tbEnv.executeSql("INSERT INTO t_enterprise_" + j +
//                        "(eid,name,credit_no,category_new,regist_capi,actual_capi,status)" +
//                        " values('" + eid + "','" + name + "','" + credit_no + "'," + category_new + "," +
//                        regist_capi + "," + actual_capi + "," + status + ")");
            }

            statementSet.execute();
        }

        // big data
        // t_enterprise <-> t_address
        List<String> citys = Arrays.asList("北京","天津","上海","广州","深圳","杭州","成都","武汉","重庆","昆明","苏州","南京");
        List<String> codes = Arrays.asList("111","222","333","444","555","666","777","888","999","1000","1100","1200");
        for (int i = 0; i < 1000; i ++) {
            boolean submit = i % 10 == 9 ? true : false;

            String eid = UUID.randomUUID().toString();
            String name = "name-" + eid;
            String credit_no = "credit-" + eid;
            String address = citys.get(i % citys.size());
            String code = codes.get(i % codes.size());
            int category_new = i % 2 == 1 ? 115601 : 115602;
            int regist_capi = 100 + i % 100;
            int actual_capi = regist_capi / 2;
            int status = i % 10 == 1 ? 1 : 0;

            statementSet.addInsertSql("INSERT INTO t_enterprise" +
                    "(eid,name,credit_no,category_new,regist_capi,actual_capi,status)" +
                    " values('" + eid + "','" + name + "','" + credit_no + "'," + category_new + "," +
                    regist_capi + "," + actual_capi + "," + status + ")");

            statementSet.addInsertSql("INSERT INTO t_address(eid,address,postcode,u_tags)" +
                    " values('" + eid + "','" + address + "','" + code + "'," + status + ")");

            if (submit) {
                statementSet.execute();
            }
        }

        // little data
        // t_enterprise <-> t_address
        for (int i = 0; i < 20; i++) {
            String eid = UUID.randomUUID().toString();
            String name = "name-" + eid;
            String credit_no = "credit-" + eid;
            String address = citys.get(i % citys.size());
            String code = codes.get(i % codes.size());
            int category_new = i % 2 == 1 ? 115601 : 115602;
            int regist_capi = 100 + i % 100;
            int actual_capi = regist_capi / 2;
            int status = i % 10 == 1 ? 1 : 0;

            statementSet.addInsertSql("INSERT INTO t_enterprise" +
                    "(eid,name,credit_no,category_new,regist_capi,actual_capi,status)" +
                    " values('" + eid + "','" + name + "','" + credit_no + "'," + category_new + "," +
                    regist_capi + "," + actual_capi + "," + status + ")");

            statementSet.addInsertSql("INSERT INTO t_address(eid,address,postcode,u_tags)" +
                    " values('" + eid + "','" + address + "','" + code + "'," + status + ")");
        }
        statementSet.execute();


        //TableEnvironment 在executeXXX时已异步执行任务，不需要再使用.execute()执行
//        tbEnv.execute("JdbcTest");
    }
}
